﻿using System;
using System.Collections.Generic;
using System.Threading;

namespace ActiveDirectoryUtilities.CommonLibrary
{
	public class WorkItemQueue : IDisposable
	{
		private class WorkItem
		{
			//Ensure uniqueness of the work item in case delgate and state are identical
			private Guid _uid;
			public WaitCallback CallBack;
			public object State;

			public WorkItem()
			{
				_uid = Guid.NewGuid();
			}
		}

		private long _maxWorkItemsConcurrency, _workItemsSubmitted, _workItemsComplete;
		private AutoResetEvent _are;
		private Queue<WorkItem> _pendingWorkItems = new Queue<WorkItem>();
		private Dictionary<WorkItem, Thread> _runningWorkItems = new Dictionary<WorkItem, Thread>();
		private object _enqueueDequequeLock = new object()
			, _manageRunningItemsDictionaryLock = new object();

		/// <summary>
		/// Initializes the WorkItemQueue class.
		/// </summary>
		/// <param name="maxWorkItemConcurrency">Set the inital maximum number of items that can be executed concurrently.
		/// Note:  If the value is less than a minimum of one 1 or greater than a maximum of ThreadPool.GetMaxThreads, it is set to the respective bound.</param>
		public WorkItemQueue(long maxWorkItemConcurrency)
		{
			int maxThreadPoolThreads, completionPortThreads;
			ThreadPool.GetMaxThreads(out maxThreadPoolThreads, out completionPortThreads);
			if (maxWorkItemConcurrency < 1)
			{
				_maxWorkItemsConcurrency = 1;
			}
			else if (maxWorkItemConcurrency > (long)maxThreadPoolThreads)
			{
				_maxWorkItemsConcurrency = (long)maxThreadPoolThreads;
			}
			else
			{
				_maxWorkItemsConcurrency = maxWorkItemConcurrency;
			}

			_are = new AutoResetEvent(false);
		}

		/// <summary>
		/// Add a work item to the queue for execution.
		/// Note:  A using statement for "System.Threading" must be added to expose the WaitCallback delegate.
		/// </summary>
		/// <param name="callback">A WaitCallback delegate pointing to the method to execute.</param>
		/// <param name="state">A set of parameters cast to an object for the method to accept.  The method must know how to translate the object passed.</param>
		public void EnqueueWorkItem(WaitCallback callback, object state)
		{
			WorkItem wi = new WorkItem();
			wi.CallBack = callback;
			wi.State = state;
			lock (_enqueueDequequeLock)
			{
				if (_pendingWorkItems.Count == 0)
				{
					_are.Reset();
				}
				_pendingWorkItems.Enqueue(wi);
				Interlocked.Increment(ref _workItemsSubmitted);
			}
			managePendingWorkItems();
		}

		private void managePendingWorkItems()
		{
			lock (_manageRunningItemsDictionaryLock)
			{
				lock (_enqueueDequequeLock)
				{
					//re-validate running work items, to ensure that this was not changed on a separate thread
					//	while waiting for a writer lock
					//if (Interlocked.Read(ref _runningWorkItems) < _maxWorkItems)
					while ((_runningWorkItems.Count < _maxWorkItemsConcurrency)
						&& (_pendingWorkItems.Count > 0))
					{
						WorkItem wi = _pendingWorkItems.Dequeue();
						//Interlocked.Increment(ref _runningWorkItems);
						_runningWorkItems.Add(wi, null);
						ThreadPool.QueueUserWorkItem(new WaitCallback(doWork), wi);
					}

					if ((_pendingWorkItems.Count == 0) && (_runningWorkItems.Count == 0))
					{
						Interlocked.Exchange(ref _workItemsSubmitted, 0);
						Interlocked.Exchange(ref _workItemsComplete, 0);
						_are.Set();
					}
				}
			}
		}

		private void doWork(object state)
		{
			WorkItem wi = (WorkItem)state;
			if (!isThreadTracked(wi))
			{
				return;
			}

			try
			{
				//Execute the actual request
				wi.CallBack(wi.State);
			}
			finally
			{
				//work item is complete, don't bother tracking it any more.
				//	Ensure work item is always removed, even if it is crashed.
				lock (_manageRunningItemsDictionaryLock)
				{
					_runningWorkItems.Remove(wi);
					Interlocked.Increment(ref _workItemsComplete);
					System.Diagnostics.Debug.WriteLine(String.Format("Exiting work item.  Threads Running {0}:", _runningWorkItems.Count));

					//If we experience ANY errors, check to make sure that the write locks are released
					//	and check if there any other items to run
				}
				managePendingWorkItems();
			}
		}

		private bool isThreadTracked(WorkItem wi)
		{
			lock (_manageRunningItemsDictionaryLock)
			{
				bool isItemAddedToRunningQueue = false;
				try
				{
					//Track the thread
					_runningWorkItems[wi] = Thread.CurrentThread;
					isItemAddedToRunningQueue = true;	//This is only true IF we can add the thread info to the Dictionary (other wise exception will skip this).
					System.Diagnostics.Debug.WriteLine(String.Format("Entering new work item.  Threads running:  {0}", _runningWorkItems.Count));
				}
				finally
				{
					//if we can't add the thread to the key, remove the running item and bail.
					if ((_runningWorkItems.ContainsKey(wi)) && (!isItemAddedToRunningQueue))
					{
						_runningWorkItems.Remove(wi);
					}
				}

				//We couldn't track the thread because of some issue, abort before we lose track.
				//	Better not to run the request than to run it without managing it.
				return isItemAddedToRunningQueue;
			}
		}

		/// <summary>
		/// The number of items still waiting to be added to the execution queue.
		/// </summary>
		public int PendingItems
		{
			get
			{
				lock (_enqueueDequequeLock)
				{
					return _pendingWorkItems.Count;
				}
			}
		}

		/// <summary>
		/// The number of items currently in the active execution state.
		/// </summary>
		public long RunningItems
		{
			get
			{
				//return Interlocked.Read(ref _runningWorkItems);
				lock (_manageRunningItemsDictionaryLock)
				{
					return _runningWorkItems.Count;
				}
			}
		}

		/// <summary>
		/// The maximum number of active items in the queue.
		/// Note:  This will not accept values less than 1 and greater than the ThreadPool.GetMaxThreads.
		/// </summary>
		public long MaxConcurrentWorkItems
		{
			get { return Interlocked.Read(ref _maxWorkItemsConcurrency); }
			set
			{
				int maxThreadPoolThreads, completionPortThreads;
				ThreadPool.GetMaxThreads(out maxThreadPoolThreads, out completionPortThreads);
				if ((value > 0) && (value <= (long)maxThreadPoolThreads))
				{
					Interlocked.Exchange(ref _maxWorkItemsConcurrency, value);
					managePendingWorkItems();
				}
			}
		}

		/// <summary>
		/// Block calling thread until all pending and active work items are complete.
		/// </summary>
		public void WaitAllWorkItems()
		{
			if ((this.RunningItems > 0) || (this.PendingItems > 0))
			{
				_are.WaitOne();
			}
		}

		#region IDisposable Members

		//TODO:  Abort all running threads.
		public void Dispose() { }

		#endregion
	}
}
